//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// buffer_pool_manager.cpp
//
// Identification: src/buffer/buffer_pool_manager.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "buffer/buffer_pool_manager.h"

#include "../../tools/sqllogictest/parser.h"
#include "common/exception.h"
#include "common/macros.h"
#include "storage/page/page_guard.h"

namespace bustub {

BufferPoolManager::BufferPoolManager(size_t pool_size, DiskManager *disk_manager, size_t replacer_k,
                                     LogManager *log_manager)
    : pool_size_(pool_size), disk_scheduler_(std::make_unique<DiskScheduler>(disk_manager)), log_manager_(log_manager) {
  // we allocate a consecutive memory space for the buffer pool
  pages_ = new Page[pool_size_];
  replacer_ = std::make_unique<LRUKReplacer>(pool_size, replacer_k);

  // Initially, every page is in the free list.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }
}

BufferPoolManager::~BufferPoolManager() { delete[] pages_; }

/**
 * 1. 先去freelist里找是否有frame是空的
 * 2. 如果有空的，就获得一张新页面号(AllocatePage())
 * 3. 把这个页放进对应的frame中，并且设置为不可驱逐
 * 4. 记录操作
 * 5. 返回这个页面
 * @param page_id
 * @return
 */
auto BufferPoolManager::NewPage(page_id_t *page_id) -> Page * {
  std::scoped_lock<std::mutex> lock(latch_);
  // freelist里有空的frame
  if (!free_list_.empty()) {
    auto frame_id = free_list_.front();
    free_list_.pop_front();
    page_id_t new_page_id = AllocatePage();
    *page_id = new_page_id;
    pages_[frame_id].ResetMemory();
    replacer_->RecordAccess(frame_id, AccessType::Unknown);
    replacer_->SetEvictable(frame_id, false);
    page_table_.insert({new_page_id, frame_id});
    pages_[frame_id].pin_count_++;
    pages_[frame_id].page_id_ = new_page_id;
    pages_[frame_id].is_dirty_ = true;
    return &pages_[frame_id];
  }
  // freelist中没有空的frame，要去raplacer里找个新的frame
  frame_id_t frame_id;
  if (replacer_->Evict(&frame_id)) {
    if (pages_[frame_id].IsDirty()) {
      auto promise = disk_scheduler_->CreatePromise();
      auto future = promise.get_future();
      auto page_ = pages_ + frame_id;
      disk_scheduler_->Schedule(DiskRequest{true, page_->GetData(), page_->GetPageId(), std::move(promise)});
      future.get();
      pages_[frame_id].is_dirty_ = false;
    }
    pages_[frame_id].ResetMemory();
    auto evict_page_id = pages_[frame_id].page_id_;
    page_table_.erase(evict_page_id);
    auto new_page_id = AllocatePage();
    *page_id = new_page_id;
    replacer_->RecordAccess(frame_id, AccessType::Unknown);
    replacer_->SetEvictable(frame_id, false);
    page_table_.insert({new_page_id, frame_id});
    pages_[frame_id].pin_count_++;
    pages_[frame_id].is_dirty_ = true;
    pages_[frame_id].page_id_ = new_page_id;
    return &pages_[frame_id];
  }
  return nullptr;
}

auto BufferPoolManager::FetchPage(page_id_t page_id, [[maybe_unused]] AccessType access_type) -> Page * {
  std::scoped_lock<std::mutex> lock(latch_);
  if (page_table_.find(page_id) != page_table_.end()) {
    frame_id_t frame_id = page_table_[page_id];
    pages_[frame_id].pin_count_++;
    replacer_->SetEvictable(frame_id, false);
    replacer_->RecordAccess(frame_id, AccessType::Unknown);
    return &pages_[frame_id];
  }
  if (!free_list_.empty()) {
    auto frame_id = free_list_.front();
    free_list_.pop_front();
    pages_[frame_id].ResetMemory();
    page_table_[page_id] = frame_id;
    pages_[frame_id].pin_count_++;
    pages_[frame_id].page_id_ = page_id;
    auto promise = disk_scheduler_->CreatePromise();
    auto future = promise.get_future();
    auto page_ = pages_ + frame_id;
    disk_scheduler_->Schedule(DiskRequest{false, page_->GetData(), page_->GetPageId(), std::move(promise)});
    future.get();
    replacer_->RecordAccess(frame_id, AccessType::Unknown);
    return &pages_[frame_id];
  }
  frame_id_t frame_id;
  if (replacer_->Evict(&frame_id)) {
    if (pages_[frame_id].IsDirty()) {
      auto promise = disk_scheduler_->CreatePromise();
      auto future = promise.get_future();
      auto page_ = pages_ + frame_id;
      disk_scheduler_->Schedule(DiskRequest{true, page_->GetData(), page_->GetPageId(), std::move(promise)});
      future.get();
      pages_[frame_id].is_dirty_ = false;
    }
    pages_[frame_id].ResetMemory();
    auto evict_page_id = pages_[frame_id].page_id_;
    page_table_.erase(evict_page_id);
    page_table_[page_id] = frame_id;
    pages_[frame_id].page_id_ = page_id;
    pages_[frame_id].pin_count_++;
    auto promise = disk_scheduler_->CreatePromise();
    auto future = promise.get_future();
    disk_scheduler_->Schedule(DiskRequest{false, pages_[frame_id].GetData(), page_id, std::move(promise)});
    future.get();
    replacer_->RecordAccess(frame_id, AccessType::Unknown);
    return &pages_[frame_id];
  }
  return nullptr;
}

auto BufferPoolManager::UnpinPage(page_id_t page_id, bool is_dirty, [[maybe_unused]] AccessType access_type) -> bool {
  std::scoped_lock<std::mutex> lock(latch_);
  auto it = page_table_.find(page_id);

  // not in buffer pool, return false
  if (it == page_table_.end() || pages_[it->second].pin_count_ <= 0) {
    return false;
  }
  frame_id_t frame_id = it->second;
  pages_[frame_id].pin_count_--;
  if (!pages_[frame_id].GetPinCount()) {
    replacer_->SetEvictable(frame_id, true);
  }
  return true;
}

auto BufferPoolManager::FlushPage(page_id_t page_id) -> bool {
  std::scoped_lock<std::mutex> lock(latch_);
  if (page_id == INVALID_PAGE_ID) {
    return false;
  }
  if (page_table_.find(page_id) == page_table_.end()) {
    return false;
  }
  auto frame_id = page_table_[page_id];
  auto promise = disk_scheduler_->CreatePromise();
  auto future = promise.get_future();
  disk_scheduler_->Schedule(
      DiskRequest{true, pages_[frame_id].GetData(), pages_[frame_id].GetPageId(), std::move(promise)});
  future.get();
  pages_[frame_id].is_dirty_ = false;
  return true;
}

void BufferPoolManager::FlushAllPages() {
  std::scoped_lock<std::mutex> lock(latch_);
  for (auto &it : page_table_) {
    FlushPage(it.first);
  }
}

auto BufferPoolManager::DeletePage(page_id_t page_id) -> bool {
  std::scoped_lock<std::mutex> lock(latch_);
  if (page_table_.find(page_id) == page_table_.end()) {
    return true;
  }
  frame_id_t frame_id = page_table_[page_id];
  if (pages_[frame_id].GetPinCount() > 0) {
    return false;
  }
  page_table_.erase(page_id);
  free_list_.push_back(frame_id);
  DeallocatePage(page_id);
  pages_[frame_id].ResetMemory();
  pages_[frame_id].pin_count_ = 0;
  pages_[frame_id].is_dirty_ = false;
  pages_[frame_id].page_id_ = INVALID_PAGE_ID;
  return false;
}

auto BufferPoolManager::AllocatePage() -> page_id_t { return next_page_id_++; }

auto BufferPoolManager::FetchPageBasic(page_id_t page_id) -> BasicPageGuard {
  auto page = FetchPage(page_id, AccessType::Unknown);
  return {this, page};
}

auto BufferPoolManager::FetchPageRead(page_id_t page_id) -> ReadPageGuard {
  auto page = FetchPage(page_id, AccessType::Unknown);
  if (page != nullptr) {
    page->RLatch();
  }
  return {this, page};
}

auto BufferPoolManager::FetchPageWrite(page_id_t page_id) -> WritePageGuard {
  auto page = FetchPage(page_id, AccessType::Unknown);
  if (page != nullptr) {
    page->WLatch();
  }
  return {this, page};
}

auto BufferPoolManager::NewPageGuarded(page_id_t *page_id) -> BasicPageGuard {
  auto page = NewPage(page_id);
  return {this, page};
}

}  // namespace bustub
